package com.fxkj.cloud.drms.domain.linkage;

import com.fxkj.cloud.dbs.common.event.stream.RemoteSceneLinkageEnableStatusEvent;
import com.fxkj.cloud.dbs.common.event.StreamBridgeConstants;
import com.fxkj.cloud.dbs.common.model.DeviceAttributeKey;
import com.fxkj.cloud.dbs.common.model.DeviceBasicInfo;
import com.fxkj.cloud.dbs.common.model.command.DeviceKvSet;
import com.fxkj.cloud.dbs.common.model.command.DeviceQuery;
import com.fxkj.cloud.dbs.common.model.kv.TsKvEntry;
import com.fxkj.cloud.dbs.common.model.linkage.ExecuteActionSceneLinkageDetail;
import com.fxkj.cloud.dbs.common.model.linkage.FxExecuteAction;
import com.fxkj.cloud.dbs.common.model.linkage.FxExecuteActionSceneLinkageType;
import com.fxkj.cloud.dbs.common.model.linkage.FxSceneLinkage;
import com.fxkj.cloud.drms.domain.device.repository.DeviceRepository;
import com.fxkj.cloud.drms.domain.linkage.event.SceneLinkageExecuteEvent;
import com.fxkj.cloud.drms.domain.mqtt.CommandDomain;
import com.fxkj.cloud.drms.domain.mqtt.CommandKv;
import com.fxkj.cloud.drms.domain.mqtt.QueryDomain;
import com.fxkj.cloud.drms.domain.mqtt.service.SenderService;
import com.fxkj.cloud.drms.domain.ts.service.TimeseriesService;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.MoreExecutors;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.context.ApplicationEventPublisher;


import java.util.*;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.function.Consumer;
import java.util.stream.Collectors;

/**
 * <p>description: 物联场景处理器</p>
 * @author ww
 * @date 2025/3/28 13:42
 */
public class FxLinkageProcessor implements FxProcessor {
    private final Logger log = LoggerFactory.getLogger(FxLinkageProcessor.class);
    private final UUID sceneId;
    private final AnyMatchFilterChain triggerFilterChain;
    private final AnyMatchFilterChain executeConditionFilterChain;
    private final List<FxExecuteAction> executeActions;
    private final TimeseriesService timeseriesService;
    private final SenderService sender;
    private final DeviceRepository deviceRepository;
    private final ApplicationEventPublisher eventPublisher;
    private final StreamBridge streamBridge;

    private volatile boolean isDestroying = false;
    private final List<ListenableFuture<?>> pendingFutures = new CopyOnWriteArrayList<>();

    public FxLinkageProcessor(FxSceneLinkage fxSceneLinkage, TimeseriesService timeseriesService, SenderService sender, DeviceRepository deviceRepository, ApplicationEventPublisher eventPublisher, StreamBridge streamBridge) {
        this.sceneId = fxSceneLinkage.getId();
        this.triggerFilterChain = new FxTriggerFilterChain(fxSceneLinkage.getTriggers());
        this.sender = sender;
        this.deviceRepository = deviceRepository;
        this.eventPublisher = eventPublisher;
        this.streamBridge = streamBridge;
        this.executeConditionFilterChain = new FxExecuteConditionFilterChain(fxSceneLinkage.getExecuteConditions(), timeseriesService);
        this.executeActions = fxSceneLinkage.getActions();
        this.timeseriesService = timeseriesService;
    }

    @Override
    public void processor(UUID deviceId, List<TsKvEntry> deviceKvs) {
        //如果正在销毁,则不处理
        if (isDestroying) {
            log.warn("处理器正在销毁,拒绝接收新任务");
            return;
        }
        // 判断触发条件
        Boolean trigger = triggerFilterChain.anyMatch(deviceId, deviceKvs);
        if (!trigger) {
            return;
        }
        // 判断执行条件
        Boolean meet = executeConditionFilterChain.anyMatch(deviceId, deviceKvs);
        if (!meet) {
            return;
        }
        // 执行动作
        ListenableFuture<Object> result = execute(deviceKvs);
        pendingFutures.add(result);
    }

    @Override
    public ListenableFuture<List<Object>> destroy() {
        // 1. 停止任务 修改当前处理器状态为 destroying, 阻止新任务的添加
        isDestroying = true;
        // 2. 等待任务执行完成, 然后销毁处理器
        // 3. 外部监听销毁处理器, 移除处理器缓存
        return Futures.allAsList(pendingFutures);

    }

    @Override
    public ListenableFuture<Object> execute(List<TsKvEntry> deviceKvs) {
        //属性下发任务
        List<ListenableFuture<List<DeviceKvSet>>> commandFutures = new ArrayList<>();
        //属性上报任务 属性查询
        List<ListenableFuture<List<DeviceQuery>>> queryFutures = new ArrayList<>();
        //场景联动任务
        List<ListenableFuture<Void>> sceneLinkageFutures = new ArrayList<>();

        Set<UUID> deviceIds = executeActions.stream().map(FxExecuteAction::getDeviceIds).flatMap(Collection::stream).collect(Collectors.toSet());
        Map<UUID, DeviceBasicInfo> deviceBasicInfoMap = deviceRepository.getDeviceBasicInfoByDeviceIds(deviceIds);
        /**
         * TODO 这里需要优化，需要根据设备id和属性key进行分组，然后并行查询
         * 目前是串行查询，会导致性能瓶颈
         */
        for (FxExecuteAction executeAction : executeActions) {
            switch (executeAction.getExecuteActionType()) {
                //属性下发
                case DEVICE_LINKAGE_ATTRIBUTE_DOWNLOAD -> {
                    List<String> attributeFilterKeys = executeAction.getAttributeFilterKeys();
                    Set<UUID> deviceIdAgg = executeAction.getDeviceFilter().getDeviceIdAgg();
                    for (UUID deviceId : deviceIdAgg) {
                        ListenableFuture<List<DeviceKvSet>> actionFuture = Futures.transformAsync(
                                timeseriesService.queryLastValueForString(deviceId, attributeFilterKeys),
                                result -> executeAction.executeCommand(result, deviceKvs),
                                MoreExecutors.directExecutor());
                        commandFutures.add(actionFuture);
                    }
                }
                //属性上报
                case DEVICE_LINKAGE_ATTRIBUTE_UPLOAD -> {
                    List<String> attributeFilterKeys = executeAction.getAttributeFilterKeys();
                    Set<UUID> deviceIdAgg = executeAction.getDeviceFilter().getDeviceIdAgg();
                    for (UUID deviceId : deviceIdAgg) {
                        ListenableFuture<List<DeviceQuery>> actionFuture = Futures.transformAsync(
                                timeseriesService.queryLastValueForString(deviceId, attributeFilterKeys),
                                result -> executeAction.executeQuery(result, deviceKvs),
                                MoreExecutors.directExecutor());
                        queryFutures.add(actionFuture);
                    }
                }
                //场景联动
                case DEVICE_SCENE_LINKAGE_LINKAGE -> {
                    ExecuteActionSceneLinkageDetail detail = executeAction.getExecuteActionSceneLinkageDetail();
                    if (null == detail) {
                        break;
                    }
                    FxExecuteActionSceneLinkageType executeType = detail.getExecuteType();
                    //执行
                    if (executeType.equals(FxExecuteActionSceneLinkageType.EXECUTE)) {
                        //推送 场景联动执行事件
                        sceneLinkageFutures.add(Futures.submit(() -> eventPublisher.publishEvent(
                                new SceneLinkageExecuteEvent(detail.getSceneLinkageId(), deviceKvs)
                        ), MoreExecutors.directExecutor()));
                    }
                    //推送 场景联动启动停止事件
                    else if (executeType.equals(FxExecuteActionSceneLinkageType.STOP) || executeType.equals(FxExecuteActionSceneLinkageType.START)) {
                        sceneLinkageFutures.add(Futures.submit(() -> {
                                    streamBridge.send(
                                            StreamBridgeConstants.TOPIC.SCENE_LINKAGE_ENABLE_STATUS,
                                            new RemoteSceneLinkageEnableStatusEvent(detail.getSceneLinkageId(), executeType));
                                    return null;
                                },
                                MoreExecutors.directExecutor()));
                    }
                }
            }
        }
        //处理命令下发
        ListenableFuture<List<DeviceKvSet>> commandTransformFuture = transform(commandFutures, result -> sendCommand(result, deviceBasicInfoMap));

        //处理属性查询
        ListenableFuture<List<DeviceQuery>> queryTransformFuture = transform(queryFutures, result -> sendQuery(result, deviceBasicInfoMap));

        //处理场景联动
        ListenableFuture<List<Void>> sceneLinkageTransformFuture = Futures.allAsList(sceneLinkageFutures);

        return Futures.transform(
                Futures.allAsList(commandTransformFuture, queryTransformFuture, sceneLinkageTransformFuture),
                input -> input.stream().
                        flatMap(Collection::stream).
                        toList(),
                MoreExecutors.directExecutor()
        );
    }

    private void sendCommand(List<DeviceKvSet> result, Map<UUID, DeviceBasicInfo> deviceBasicInfoMap) {
        result.forEach(deviceKv -> {
            DeviceBasicInfo deviceBasicInfo = deviceBasicInfoMap.get(deviceKv.deviceId());
            if (null == deviceBasicInfo) {
                return;
            }
            sender.send(CommandDomain.create(deviceBasicInfo, deviceKv.kvs().stream().map(kv -> new CommandKv(DeviceAttributeKey.create(kv.attributeCode()), kv.value())).toList(), null));
        });
    }

    private void sendQuery(List<DeviceQuery> result, Map<UUID, DeviceBasicInfo> deviceBasicInfoMap) {
        result.forEach(query -> {
            DeviceBasicInfo deviceBasicInfo = deviceBasicInfoMap.get(query.getDeviceId());
            if (null == deviceBasicInfo) {
                return;
            }
            sender.query(QueryDomain.create(deviceBasicInfo, query.getQueryKeys()));
        });
    }

    public <Type> ListenableFuture<List<Type>> transform(List<ListenableFuture<List<Type>>> commandFutures, Consumer<List<Type>> consumer) {
        ListenableFuture<List<Type>> commandTransformFuture = Futures.transform(
                Futures.allAsList(commandFutures),
                input -> input.stream().
                        flatMap(Collection::stream).
                        toList(),
                MoreExecutors.directExecutor()
        );
        Futures.addCallback(commandTransformFuture, new FutureCallback<>() {
            @Override
            public void onSuccess(List<Type> result) {
                consumer.accept(result);
            }

            @Override
            public void onFailure(Throwable t) {

            }
        }, MoreExecutors.directExecutor());
        return commandTransformFuture;
    }

    public UUID getSceneId() {
        return sceneId;
    }
}
